

package com.hazelcast.client.map.impl.querycache.subscriber;

import com.hazelcast.client.impl.protocol.ClientMessage;
import com.hazelcast.client.impl.protocol.codec.ContinuousQueryMadePublishableCodec;
import com.hazelcast.client.impl.protocol.codec.ContinuousQueryPublisherCreateCodec;
import com.hazelcast.client.impl.protocol.codec.ContinuousQueryPublisherCreateWithValueCodec;
import com.hazelcast.internal.serialization.Data;
import com.hazelcast.map.impl.querycache.InvokerWrapper;
import com.hazelcast.map.impl.querycache.accumulator.AccumulatorInfo;
import com.hazelcast.map.impl.querycache.subscriber.AbstractQueryCacheEndToEndConstructor;
import com.hazelcast.map.impl.querycache.subscriber.InternalQueryCache;
import com.hazelcast.map.impl.querycache.subscriber.QueryCacheEndToEndConstructor;
import com.hazelcast.map.impl.querycache.subscriber.QueryCacheRequest;

import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * Client-side implementation of {@code QueryCacheEndToEndConstructor}.
 *
 * @see QueryCacheEndToEndConstructor
 */
public class ClientQueryCacheEndToEndConstructor extends AbstractQueryCacheEndToEndConstructor {

    public ClientQueryCacheEndToEndConstructor(QueryCacheRequest request) {
        super(request);
    }

    private static void prepopulate(InternalQueryCache queryCache, Collection<Map.Entry<Data, Data>> result) {
        queryCache.prepopulate(result.iterator());
    }

    private static void prepopulate(InternalQueryCache queryCache, List<Data> result) {
        Iterator<Map.Entry<Data, Data>> iterator = new CachedEntryIterator(result.iterator());
        queryCache.prepopulate(iterator);
    }

    @Override
    public void createPublisherAccumulator(AccumulatorInfo info, boolean urgent) throws Exception {
        ClientMessage publisherCreateMessage = newPublisherCreateMessage(info);

        InvokerWrapper invokerWrapper = context.getInvokerWrapper();
        ClientMessage response = (ClientMessage) invokerWrapper.invoke(publisherCreateMessage, urgent);

        if (info.isIncludeValue()) {
            Collection<Map.Entry<Data, Data>> result = ContinuousQueryPublisherCreateWithValueCodec.decodeResponse(response);
            prepopulate(queryCache, result);
        } else {
            List<Data> result = ContinuousQueryPublisherCreateCodec.decodeResponse(response);
            prepopulate(queryCache, result);
        }


        if (info.isPopulate()) {
            madePublishable(info.getMapName(), info.getCacheId(), urgent);
            info.setPublishable(true);
        }
    }

    private ClientMessage newPublisherCreateMessage(AccumulatorInfo info) {
        Data dataPredicate = context.getSerializationService().toData(info.getPredicate());

        if (info.isIncludeValue()) {
            return ContinuousQueryPublisherCreateWithValueCodec.encodeRequest(info.getMapName(), info.getCacheId(), dataPredicate, info.getBatchSize(), info.getBufferSize(), info.getDelaySeconds(), info.isPopulate(), info.isCoalesce());

        }

        return ContinuousQueryPublisherCreateCodec.encodeRequest(info.getMapName(), info.getCacheId(), dataPredicate, info.getBatchSize(), info.getBufferSize(), info.getDelaySeconds(), info.isPopulate(), info.isCoalesce());
    }

    private void madePublishable(String mapName, String cacheName, boolean urgent) throws Exception {
        ClientMessage request = ContinuousQueryMadePublishableCodec.encodeRequest(mapName, cacheName);
        context.getInvokerWrapper().invokeOnAllPartitions(request, urgent);
    }

    // Adapts an Iterator<List<Data>> of keys as an Iterator<Map.Entry<Data, Data>> of KVs with null value
    private static final class CachedEntryIterator implements Iterator<Map.Entry<Data, Data>> {
        private final Iterator<Data> keyIterator;
        private final MutableEntry mutableEntry;

        CachedEntryIterator(Iterator<Data> keyIterator) {
            this.keyIterator = keyIterator;
            this.mutableEntry = new MutableEntry();
        }

        @Override
        public boolean hasNext() {
            return keyIterator.hasNext();
        }

        @Override
        public Map.Entry<Data, Data> next() {
            return mutableEntry.setKey(keyIterator.next());
        }
    }

    private static final class MutableEntry implements Map.Entry<Data, Data> {
        private Data key;

        MutableEntry() {
        }

        @Override
        public Data getKey() {
            return key;
        }

        public MutableEntry setKey(Data key) {
            this.key = key;
            return this;
        }

        @Override
        public Data getValue() {
            return null;
        }

        @Override
        public Data setValue(Data value) {
            throw new UnsupportedOperationException();
        }
    }
}
